package com.lt.lumentracks.task;

import com.lt.lumentracks.constant.RedisConstant;
import com.lt.lumentracks.entity.po.ConversationUser;
import com.lt.lumentracks.service.IConversationUserService;
import jakarta.annotation.PreDestroy;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;


// 定时任务：固定时间刷新一次用户最后阅读消息
@Component
@Slf4j
public class LastMsgFlushTask {

    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private IConversationUserService iConversationUserService;

    private static final String REDIS_KEY_PATTERN = RedisConstant.USER_LAST_READ_MSG_KEY + "*"; // 匹配规则
    private static final int BATCH_SIZE = 1_000; // 批量处理数量

    // 读取 lua 文件内容
    private static final DefaultRedisScript<Long> CAS_DEL_SCRIPT;
    static {
        CAS_DEL_SCRIPT = new DefaultRedisScript<>();
        CAS_DEL_SCRIPT.setLocation(new ClassPathResource("lua/lastMsgCasDel.lua"));
        CAS_DEL_SCRIPT.setResultType(Long.class);
    }

    private final AtomicBoolean shutdown = new AtomicBoolean(false); // 停机标记

    @Setter
    private volatile boolean taskStopped = false; // 标识任务是否被手动停止
    private static final Duration FIXED_DELAY = Duration.ofSeconds(5);


    // 定时任务
    public void flush() {
        // 优雅停机标记
        if (shutdown.get()) {
            return; // 如果应用正在关闭，直接返回，不处理新数据
        }

        StopWatch sw = new StopWatch();
        sw.start();

        // 0. 使用 SCAN 命令分批扫描，而不使用keys命令，避免 KEYS * 阻塞。
        try (Cursor<String> cursor = redisTemplate.scan(ScanOptions.scanOptions()
                .match(REDIS_KEY_PATTERN) // 匹配模式
                .count(BATCH_SIZE) // 游标分批
                .build()
        )) {

            // 1. 批量获取 key
            List<String> keys = new ArrayList<>(BATCH_SIZE);
            while (cursor.hasNext() && keys.size() < BATCH_SIZE) {
                keys.add(cursor.next());
            }
            if (keys.isEmpty()) {
                return;
            }
            log.info("开始刷新用户最后阅读消息任务...");

            // 2. 批量获取 value
            List<String> values = redisTemplate.opsForValue().multiGet(keys);
            if (CollectionUtils.isEmpty(values)) {
                return;
            }

            // 3. 组装 MySQL 参数，同时记录 <key,expectedValue>
            List<ConversationUser> cuList = new ArrayList<>(keys.size());
            Map<String, String> key2value = new HashMap<>(keys.size());
            for (int i = 0; i < keys.size(); i++) {
                String key = keys.get(i);
                String value = values.get(i);
                if (value == null) {
                    continue; // 值不见了，跳过且不删
                }
                String[] split = key.split(":");
                // 键 -> lt_userLastReadMsg:userId:用户ID:convId:会话ID  值 -> 最后阅读消息ID
                ConversationUser cu = new ConversationUser()
                        .setUserId(Integer.parseInt(split[2]))
                        .setConversationId(Long.parseLong(split[4]))
                        .setLastReadMessageId(Long.parseLong(value));
                cuList.add(cu);
                key2value.put(key, value);
            }

            if (cuList.isEmpty()) {
                return;
            }

            // 4. 批量写 MySQL
            iConversationUserService.batchUpdateLastMsg(cuList);

            // 5. CAS 删除 Redis（管道）
            List<Object> results = redisTemplate.executePipelined((RedisConnection connection) -> {
                StringRedisConnection stringConn = (StringRedisConnection) connection;
                key2value.forEach((k, v) ->
                        stringConn.eval(
                                CAS_DEL_SCRIPT.getScriptAsString(), // 脚本内容
                                ReturnType.INTEGER, // 脚本返回类型
                                1, // 参数数量
                                k,
                                v));
                return null;
            });

            long delOk = results.stream().mapToLong(r -> (Long) r).sum();
            sw.stop();
            log.info("刷新用户最后阅读消息任务完成, batch={}, mysqlRows={}, casDelOk={},耗时={}ms",
                    keys.size(), cuList.size(), delOk, sw.getTotalTimeMillis());
        }
    }


    // 在Bean销毁前，调用的方法
    @PreDestroy
    public void gracefulShutdown() throws InterruptedException {
        if (taskStopped) {
            return; // 任务已被手动停止，不用等待最后一次调度完成
        }
        shutdown.set(true);
        log.warn("LastMsgFlushTask 准备停机，等待最后一次调度完成...");
        Thread.sleep(FIXED_DELAY.toMillis() + 500);
    }

}
